package cn.ryanpenn.bigdata.hadoop.wordcount;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;


/**
 * 词频统计
 *
 * @author Ryan
 */
public class WordCountDriver extends Configured implements Tool {

    public int run(String[] args) throws Exception {
        if (args.length != 2) {
            System.err.printf("Usage: %s <input path> <output path>\n", this.getClass().getSimpleName());
            ToolRunner.printGenericCommandUsage(System.err);
            return -1;
        }

        // 为了让示例程序能够再win和mac上调试，在生产环境不要这样使用
        String os = "win";
        if (System.getProperty("os.name").toLowerCase().indexOf("mac") > 0)
            os = "mac";

        // 加载配置文件
        Configuration conf = new Configuration();
        conf.addResource(os + "/core-site.xml");
        conf.addResource(os + "/hdfs-site.xml");
        conf.addResource("mapred-site.xml");
        conf.addResource("yarn-site.xml");

        // 设置参数
        conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");
        conf.set("mapreduce.job.jar", "target/wordCountJob-with-dependencies.jar");
        conf.set("mapreduce.framework.name", "yarn");
        conf.set("yarn.resourcemanager.hostname", "localhost");
        conf.set("mapreduce.app-submission.cross-platform", "true");

        // 如果输出文件已经存在，则删除之前的结果
        FileSystem fs = FileSystem.get(conf);
        Path output = new Path(args[1]);
        if (fs.exists(output))
            fs.delete(output, true);
        fs.close();

        // 新建一个作业
        Job job = Job.getInstance(conf, "word count");
        job.setJarByClass(WordCountDriver.class);

        // 设置mapper类、reducer类
        job.setMapperClass(WordCountMapper.class);
        job.setReducerClass(WordCountReducer.class);
        // 注意：Combiner不一定适合所有场景，如求平均数时如果设置了Combiner可能导致结果不正确
        job.setCombinerClass(WordCountReducer.class);

        // 设置Key和Value类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        // 设置输入、输出路径（均为hdfs文件系统中的路径）
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        // 等待执行完成
        return job.waitForCompletion(true) ? 0 : 1;
    }
}
